package hbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 
 * @version 1.0
 * @date 2021/10/9 17:43
 */
@Slf4j
public class HBaseTemplate {

    private final Configuration entries;

    private volatile Connection connection;

    public HBaseTemplate(Configuration entries) {
        Assert.notNull(entries, "HBase 配置类初始化失败！");
        this.entries = entries;
        getConnection();
    }

    public Connection getConnection() {
        if (null == connection) {
            synchronized (this) {
                if (null == connection) {
                    try {
                        connection = ConnectionFactory.createConnection(entries);
                        log.info("HBase 连接建立成功！");
                    } catch (Exception e) {
                        log.error("HBase 连接建立失败！");
                        e.printStackTrace();
                    }
                }
            }
        }
        return connection;
    }


    public <T> List<T> scan(String tableName, Class<T> clazz) {
        return executeScan(tableName, new Scan(), clazz);
    }


    public <T> List<T> scan(String tableName, String rowKeyStart, String rowKeyEnd, Class<T> clazz) {
        Scan scan = new Scan();
        if (null != rowKeyStart && !rowKeyStart.isEmpty()) {
            scan.withStartRow(Bytes.toBytes(rowKeyStart));
        }
        if (null != rowKeyEnd && !rowKeyEnd.isEmpty()) {
            scan.withStopRow(Bytes.toBytes(rowKeyEnd));
        }

        return executeScan(tableName, scan, clazz);
    }


    public <T> List<T> scan(String tableName, String rowKeyPrefix, Class<T> clazz) {
        Scan scan = new Scan();
        scan.withStartRow(rowKeyPrefix.getBytes());
        scan.setFilter(new PrefixFilter(rowKeyPrefix.getBytes()));

        return executeScan(tableName, scan, clazz);
    }


    public <T> List<T> scanMoreThanRowKey(String tableName, String rowKey, Class<T> clazz) {
        Scan scan = new Scan();
        Filter filter = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(rowKey.getBytes()));
        scan.setFilter(filter);

        return executeScan(tableName, scan, clazz);
    }


    public <T> List<T> scanLessThanRowKey(String tableName, String rowKey, Class<T> clazz) {
        Scan scan = new Scan();
        Filter filter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL,
                new BinaryComparator(rowKey.getBytes()));
        scan.setFilter(filter);

        return executeScan(tableName, scan, clazz);
    }

    private <T> List<T> executeScan(String tableName, Scan scan, Class<T> clazz) {
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            ResultScanner scannerResult = table.getScanner(scan);
            List<Map<String, String>> hDataList = new ArrayList<>();
            for (Result result : scannerResult) {
                Map<String, String> map = new HashMap<>();
                if (null == result.rawCells()) {
                    log.debug("查询到记录为空!");
                    continue;
                }
                for (Cell cell : result.rawCells()) {
                    String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    map.put(colName, value);
                }
                if (!map.isEmpty()) {
                    hDataList.add(map);
                }
            }
            if (clazz.isAssignableFrom(HashMap.class)) {
                return (List<T>) hDataList;
            } else {
                return JsonUtil.parseList(JsonUtil.toJsonString(hDataList), clazz);
            }
        } catch (Exception e) {
            log.error("扫描表查询批量数据时发生异常，e:", e);
            return null;
        }
    }


    public <T> T getByRowKey(String tableName, String rowKey, Class<T> clazz) {
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            if (null == result.listCells()) {
                log.debug("查询到 rowKey={} 的结果为空!", rowKey);
                return null;
            }
            Map<String, String> map = new HashMap<>();
            for (Cell cell : result.listCells()) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                map.put(colName, value);
            }
            if (clazz.isAssignableFrom(HashMap.class)) {
                return (T) map;
            } else {
                return JsonUtil.parseObject(JsonUtil.toJsonString(map), clazz);
            }
        } catch (Exception e) {
            log.error("根据rowKey查询单条数据时发生异常，e:", e);
            return null;
        }
    }


    public <T> List<T> getBatchByRowKey(String tableName, List<String> rowKeyList, Class<T> clazz) {
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            List<Get> getList = new ArrayList<>();
            for (String rowKey : rowKeyList) {
                Get get = new Get(Bytes.toBytes(rowKey));
                getList.add(get);
            }
            Result[] results = table.get(getList);
            List<Map<String, String>> list = new ArrayList<>();
            for (Result result : results) {
                if (null == result.listCells()) {
                    log.debug("查询到的数据为空!");
                    continue;
                }
                Map<String, String> map = new HashMap<>();
                for (Cell cell : result.listCells()) {
                    String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    map.put(colName, value);
                }
                list.add(map);
            }
            if (clazz.isAssignableFrom(HashMap.class)) {
                return (List<T>) list;
            } else {
                return JsonUtil.parseList(JsonUtil.toJsonString(list), clazz);
            }
        } catch (Exception e) {
            log.error("根据rowKey查询批量数据时发生异常，e:", e);
            return null;
        }
    }


    public String getCell(String tableName, String rowKey, String columnFamily, String column) {
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
            Result result = table.get(get);
            Cell[] cells = result.rawCells();
            if (null == cells || cells.length == 0) {
                log.debug("未查询到值![tableName={},rowKey={},columnFamily={},column={}]", tableName, rowKey, columnFamily, column);
                return null;
            }
            return Bytes.toString(cells[0].getValueArray(), cells[0].getValueOffset(), cells[0].getValueLength());
        } catch (Exception e) {
            log.error("查询列值时发生异常，e:", e);
            return null;
        }
    }


    public boolean deleteByRowKey(String tableName, String rowKey) {
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            return true;
        } catch (Exception e) {
            log.error("删除数据时发生异常，e:", e);
            return false;
        }
    }


    public int deleteByRowKeyPrefix(String tableName, String rowKeyPrefix) {
        try {
            int result = 0;
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setFilter(new PrefixFilter(Bytes.toBytes(rowKeyPrefix)));
            ResultScanner rs = table.getScanner(scan);

            List<Delete> deleteList = new ArrayList<>();
            Delete delete = null;
            for (Result r : rs) {
                delete = new Delete(r.getRow());
                deleteList.add(delete);
            }
            result = deleteList.size();
            table.delete(deleteList);
            return result;
        } catch (Exception e) {
            log.error("批量删除时发生异常，e:", e);
        }
        return 0;
    }


    public boolean insert(String tableName, String rowKey, String columnFamily, String column, String data) {
        if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(rowKey) || StringUtils.isEmpty(columnFamily) || StringUtils.isEmpty(column)) {
            log.debug("必要数据不能为空！");
            return false;
        }
        try {
            Table table = this.getConnection().getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            table.put(put);
            return true;
        } catch (Exception e) {
            log.error("插入/更新数据时发生异常，e:", e);
        }
        return false;
    }


    public boolean createTable(String tableName, String... columnFamilies) {
        try {
            Admin admin = this.getConnection().getAdmin();
            TableName tableName1 = TableName.valueOf(tableName);
            if (admin.tableExists(tableName1)) {
                log.debug("创建失败：表 {} 已存在!", tableName);
                return false;
            }
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName1);
            for (String cf : columnFamilies) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            return true;
        } catch (Exception e) {
            log.error("创建表时发生异常，e:", e);
        }
        return false;
    }


    public boolean enableTable(String tableName) {
        try {
            Admin admin = connection.getAdmin();
            TableName tableNameBean = TableName.valueOf(tableName);
            if (admin.tableExists(tableNameBean)) {
                if (!admin.isTableDisabled(tableNameBean)) {
                    log.debug("使表生效失败：表 {} 已经处于生效状态!", tableName);
                    return false;
                }
                admin.enableTable(tableNameBean);
                return true;
            } else {
                log.debug("使表生效失败：表 {} 不存在!", tableName);
                return false;
            }
        } catch (Exception e) {
            log.error("使表生效时发生异常，e:", e);
        }
        return false;
    }


    public boolean disEnableTable(String tableName) {
        try {
            Admin admin = connection.getAdmin();
            TableName tableNameBean = TableName.valueOf(tableName);
            if (admin.tableExists(tableNameBean)) {
                if (admin.isTableDisabled(tableNameBean)) {
                    log.debug("使表失效失败：表 {} 已经处于失效状态!", tableName);
                    return false;
                }
                admin.disableTable(tableNameBean);
                return true;
            } else {
                log.debug("使表失效失败：表 {} 不存在!", tableName);
                return false;
            }
        } catch (Exception e) {
            log.error("使表失效时发生异常，e:", e);
        }
        return false;
    }


    public boolean deleteTable(String tableName) {
        try {
            Admin admin = this.connection.getAdmin();
            TableName tableNameBean = TableName.valueOf(tableName);
            if (admin.tableExists(tableNameBean)) {
                if (!admin.isTableDisabled(tableNameBean)) {
                    admin.disableTable(tableNameBean);
                }
                admin.deleteTable(tableNameBean);
                return true;
            } else {
                log.debug("删除表失败：表 {} 不存在!", tableName);
                return false;
            }
        } catch (Exception e) {
            log.error("删除表时发生异常，e:", e);
        }
        return false;
    }
}
